package com.dongdong.mq;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSONObject;
import com.dongdong.pojo.TbOrder;
import com.dongdong.pojo.TbScoreCount;
import com.dongdong.pojo.TbUser;
import com.dongdong.service.OrderService;
import com.dongdong.service.ScoreService;
import com.dongdong.utils.IdWorker;
import org.apache.catalina.security.SecurityUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.shiro.SecurityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.Date;
import java.util.List;

@Component
@RocketMQMessageListener(topic = "order-topic",consumerGroup = "dondong_score_group")
public class RocketMQConsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
    @Reference
    private OrderService orderService;
    @Autowired
    private RedisTemplate redisTemplate;
    @Reference
    private ScoreService scoreService;
    @Override
    public void onMessage(String s) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt msg = list.get(0);
                String msgId = msg.getMsgId();
                try {
                    Object socre = redisTemplate.boundHashOps("socre").get(msgId);
                    if (socre == null){
                        String msgBody = new String(msg.getBody(), "utf-8");
                        TbOrder tbOrder = JSONObject.parseObject(msgBody,TbOrder.class);
                        TbScoreCount scoreCount = new TbScoreCount();
                        Double doubleValue = tbOrder.getPayment().doubleValue();
                        tbOrder.getPayment().longValue();
                        if (doubleValue <=1){
                            scoreCount.setExpendScore("1");
                            scoreCount.setTotalScore(1L);
                        }else {
                            scoreCount.setTotalScore(tbOrder.getPayment().longValue());
                            scoreCount.setExpendScore(tbOrder.getPayment().toString());
                        }
                        scoreCount.setId(new IdWorker().nextId());
                        scoreCount.setCreateTime(new Date());
                        scoreCount.setUpdateTime(new Date());
                        scoreService.insert(scoreCount);
                        // 消费成功后  将msgId作为key 存进redis
                        redisTemplate.boundHashOps("socre").put(msgId,"1");
                    }
                } catch (Exception e){
                    e.printStackTrace();
                    int reconsumeTimes = msg.getReconsumeTimes();
                    if(reconsumeTimes > 2){
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER; //若消费失败,稍后再消费
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
            }
        });
    }
}
